package com.ruoyi.flink.util.read;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import com.alibaba.fastjson.JSONArray;
import com.ruoyi.flink.util.GetBeanFromResultUtil;

/**
 * @Description:flink从mysql中读取数据
 * @ClassName: MysqlRead
 * @author syc
 * @date: 2022年5月30日 上午8:53:38 早知如此绊人心,何如当初莫相识;
 */
public class MysqlRead extends RichSourceFunction<JSONArray> {
	private static final long serialVersionUID = 1L;
	Connection connection = null;
	PreparedStatement preparedStatement = null;
	String sql;
	String url = "jdbc:mysql://127.0.0.1:3306/mytest?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false";
	String user = "root";
	String password = "19901110";

	public MysqlRead(String sql) {
		this.sql = sql;
	}

	public MysqlRead(String sql, String url, String user, String password) {
		this.sql = sql;
		this.url = url;
		this.url = user;
		this.password = password;
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		// Class.forName("com.mysql.cj.jdbc.Driver");
		String url = "jdbc:mysql://127.0.0.1:3306/mytest?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false";
		connection = DriverManager.getConnection(url, "root", "19901110");
	}

	@Override
	public void run(SourceContext<JSONArray> ctx) throws Exception {
		preparedStatement = connection.prepareStatement(sql);
		ResultSet resultSet = preparedStatement.executeQuery();
		if (resultSet != null) {
			while (resultSet.next()) {
				ctx.collect(GetBeanFromResultUtil.resultSetToJsonArry(resultSet));
			}
		}
	}
	@Override
	public void cancel() {
		try {
			if (preparedStatement != null) {
				preparedStatement.close();
			}
			if (connection != null) {
				connection.close();
			}
		} catch (SQLException throwables) {
			throwables.printStackTrace();
		}
	}

}
